# coding: UTF-8
import pandas as pd
from pyspark.sql import SQLContext
import sys
reload(sys)
sys.setdefaultencoding('utf-8')
sqlContest = SQLContext(sc)

def extractions_tmp_2(keyword,provide_id):
    sql1="""select * from games.temp_20180721 where  msg_content like '%%%(keyword)s%%' and
        provider_id =%(provide_id)d
        and city not in('北京','上海','深圳市','天津') and prov_id != 12 and status =3""" % {"keyword":keyword,"provide_id": provide_id}
    print sql1
    df=sqlContext.sql(sql1)
    df.registerTempTable("aaaaa")
    sql="""
        select
            t2.msisdn msisdn,
            t2.city city,
            t2.provider_id provider_id
        from (
            select
                t.msisdn msisdn ,
                t.city city,
                t.provider_id provider_id
            from
                aaaaa  t
            LEFT OUTER JOIN
                (SELECT  msisdn FROM history.games_zcx UNION ALL SELECT  msisdn FROM history.temp ) t1
            ON t.msisdn = t1.msisdn
            WHERE t1.msisdn is null
        ) t2
        group by
            t2.msisdn,
            t2.city,
            t2.provider_id
    """
    print sql
    df = sqlContext.sql(sql)
    print "+++++++++++++++++++++++++++++++++++++++"
    print "+++++++++++++++++++++++++++++++++++++++"
    print "+++++++++++++++++++++++++++++++++++++++"
    print "+++++++++++++++++++++++++++++++++++++++"
    print """%(keyword)s OK""" %{"keyword":keyword}
    return df

df=extractions_tmp_2("热血传奇", 1)
df.count()



